/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.scheduler;

import org.apache.log4j.Logger;

import java.util.Collection;
import java.util.LinkedList;

import jp.aist.gtrc.plus.reserve.ReserveId;
import jp.aist.gtrc.plus.reserve.ReserveInfo;
import jp.aist.gtrc.plus.reserve.ReserveManager;
import jp.aist.gtrc.plus.scheduler.job.JobManager;
import jp.aist.gtrc.plus.scheduler.job.Scheduler;
import jp.aist.gtrc.plus.scheduler.node.ExecNodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocateRequest;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeInfo;
import jp.aist.gtrc.plus.scheduler.node.NodeSorter;
import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.status.MainServer;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;
import jp.aist.gtrc.plus.scheduler.util.CollectionUtil;

abstract class DefaultSelfScheduler extends DefaultScheduler {

        protected static final Logger logger = Logger.getLogger(DefaultSelfScheduler.class);

	protected final JobManager jobMgr;
	protected final Scheduler scheduler;
	protected final NodeAllocator nodeAllocator;

	protected DefaultSelfScheduler(SchedulerOptions options) throws Exception {

		super(options);

		scheduler = Scheduler.getScheduler(options);
		nodeAllocator = getExecNodeAllocator(options);
		jobMgr = getJobManager(rsvMgr);
		assert(jobMgr != null);
		jobMgr.useStarvedJobHelper(24 * 3600);

	}

	protected NodeAllocator getExecNodeAllocator(SchedulerOptions options) throws SchedulerException {

		return new ExecNodeAllocator(new NodeSorter(options));

	}

	protected JobManager getJobManager(ReserveManager rsvMgr) throws SchedulerException {

		return new JobManager(rsvMgr);

	}

	protected abstract MainServer getMainServer() throws Exception;

	protected void runSchedule(ScheduleOrder order) throws SchedulerException{
                long before = System.currentTimeMillis();
                updateStatus(order.getMainServer());
                long after = System.currentTimeMillis();
System.err.println("runSchedule() spent " + (after-before) + " msec. in updateStatus()");

		ReserveId rsvId = order.getReserveId();
		if (rsvId != null) {
			handleReserveCancel(order);
		} else {
			if (handleWaitingTransaction()) {
				// waiting transaction was committed/aborted automatically.
				// don't schedule jobs until next turn.
				return;
			}
			handleReserveExpire(order);
			schedule(order);
		}

	}

	protected void schedule(ScheduleOrder order) throws SchedulerException{

		assert(jobMgr != null);
		assert(scheduler != null);

                logger.debug("DefaultSelfScheduler schedule()"); 

		/*
		 * Select one job to run
		 */
		Collection<JobStatus> allJobs = statusMgr.getJobs();
		Collection<JobStatus> runnableJobs = jobMgr.getRunnableJobs(allJobs);
		if (runnableJobs != null) {
			JobStatus job = scheduler.selectJob(runnableJobs);
			runOneJob(order.getMainServer(), job);
		}

	}

	protected void runOneJob(MainServer srv, JobStatus job) throws SchedulerException {

		assert(nodeAllocator != null);

		/*
		 * Select exec-nodes to use
		 */
		ReserveId rsvId = job.getReserveId();
		ReserveInfo rsvInfo = rsvMgr.getReserveInfo(rsvId);
		Collection<NodeInfo> usableNodes, nodes;
		if (rsvInfo != null) {
			usableNodes = rsvInfo.getReservedNodeInfos();
			boolean mustRestart = handleUnreservedJobs(srv, usableNodes);
			if (mustRestart) {
				// some information was changed by handleUnreservedJobs(),
				// don't call runJob(), restart scheduling
				return;
			}
		} else {
			usableNodes = nodeMgr.getUnreservedNodes();
		}

		nodes = nodeAllocator.suitable(usableNodes, new NodeAllocateRequest(job));
		if (nodes != null) {
			/*
			 * Run the job at nodes
			 */
			srv.runJob(job, nodes);
		} else {
			// no enough exec-nodes
			// ex: reserve is correct, but too many jobs were submitted
                         
		}

	}

	protected boolean handleUnreservedJobs(MainServer srv,
			Collection<NodeInfo> rsvdNodes) throws SchedulerException {

		// do nothing, override this if needed

		// false: OK to continue
		// true: some job was reran (killed), must restart scheduling cycle
		return false;

	}

	private boolean handleWaitingTransaction() throws SchedulerException {

		return rsvMgr.scheduleWaitingCommitAbort();

	}

	private void holdJobs(MainServer srv, Collection<JobStatus> jobs) {

		if (CollectionUtil.isEmpty(jobs)) {
			return;	// nothing to hold, do nothing
		}

		for (JobStatus job : jobs) {
			try {
				switch (job.getState()) {
				case Queued:
					srv.holdJob(job.getJobID());
					break;
				case Running:
					srv.rerunJob(job);
					break;
				case Held:
				case Exiting:
				case Suspended:
					break;
				default:
					assert(false);
					break;
				}
			} catch (SchedulerException e) {
				logger.info("cannot hold/rerun of " + job.getName());
			}
		}

	}

	protected void handleReserveExpire(ScheduleOrder order) throws SchedulerException{

		Collection<ReserveInfo> expiredRsv
			= new LinkedList<ReserveInfo>(rsvMgr.getExpiredReserve());
		if (expiredRsv.isEmpty()) {
			return;	// no expired reserves
		}

		Collection<JobStatus> allJobs = statusMgr.getJobs();
		Collection<JobStatus> holdJobsList = new LinkedList<JobStatus>();
		for (ReserveInfo info : expiredRsv) {
			ReserveId expireId = info.getReserveId();
			for (JobStatus job : allJobs) {
				ReserveId id = job.getReserveId();
				if ((id != null) && id.equals(expireId)) {
					holdJobsList.add(job);
				}
			}
		}
		holdJobs(order.getMainServer(), holdJobsList);

		// Remove expired reserves.
		// Their state maybe not only DONE but also CANCELED or ERROR
		for (ReserveInfo info : expiredRsv) {
			rsvMgr.remove(info);
		}

	}

	protected void handleReserveCancel(ScheduleOrder order) throws SchedulerException{

		ReserveId rsvId = order.getReserveId();
		if (rsvId == null) {
			return;
		}

		ReserveInfo info = rsvMgr.getReserveInfo(rsvId);
		if (info != null) {
			switch (info.getReserveState()) {
			case Confirmed:
			case Running:
				return;
			default:
				break;
			}
		}

		Collection<JobStatus> holdJobsList = new LinkedList<JobStatus>();

		for (JobStatus job : statusMgr.getJobs()) {
			ReserveId id = job.getReserveId();
			if ((id != null) && id.equals(rsvId)) {
				holdJobsList.add(job);
			}
		}

		holdJobs(order.getMainServer(), holdJobsList);

	}

}
